package com.gzyj.flink.gps.broadcast;

import com.gzyj.flink.driver.Driverlogininfo;
import com.gzyj.flink.gp.zdaq.TCmd020065;
import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.addtionalfnfo.JTT808_AddtionalInfo;
import com.gzyj.flink.gps.addtionalfnfo.subiao.JTT808_0x0200_0x65;
import com.gzyj.flink.utils.PointUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Date;


/**
 * @author lhj
 */
public class TCmd020065Process extends RichFlatMapFunction<GPSPosition, TCmd020065> {

    private Connection connection;
    private PreparedStatement ps;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
            ExecutionConfig.GlobalJobParameters globalJobParam = getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
            Configuration globconf = (Configuration) globalJobParam;
            this.connection = DriverManager.getConnection(globconf.getString("mysql-url", null),
                    globconf.getString("mysql-user", "root"),
                    globconf.getString("mysql-password", "123456"));

            String sql = "SELECT t.* FROM (SELECT VehId, max( LoginTime ) AS LoginTime FROM driverlogininfo GROUP BY VehId ) as a LEFT JOIN driverlogininfo t on a.VehId=t.VehId and a.LoginTime=t.LoginTime where t.VehId=?";


            this.ps = this.connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            this.ps.setFetchSize(50 * 1000);
            this.ps.setMaxRows(50 * 1000);
            this.ps.setFetchDirection(ResultSet.FETCH_REVERSE);
        } catch (Exception e) {
            System.out.println("-----------mysql get connection has exception , msg = " + e.getMessage());
        }


    }




    @Override
    public void close() throws Exception {
        try {

            if (connection != null) { //关闭连接和释放资源
                connection.close();
            }
            if (ps != null) {
                ps.close();
            }
        } catch (Exception e) {

        }
        super.close();
    }

    @Override
    public void flatMap(GPSPosition gpsPosition, Collector<TCmd020065> collector) throws Exception {
        this.ps.setString(1, gpsPosition.getVehId());
        ResultSet rs = this.ps.executeQuery();
        Driverlogininfo driverlogininfo = new Driverlogininfo();
        while (rs.next()) {
            driverlogininfo.setCheckObject(rs.getInt("CheckObject"));
            driverlogininfo.setLoginResults(rs.getInt("LoginResults"));
            driverlogininfo.setDriverName(rs.getString("DriverName"));
            driverlogininfo.setDriverNo(rs.getString("DriverNo"));
            driverlogininfo.setLoginTime(rs.getDate("LoginTime"));
            driverlogininfo.setVehId(rs.getString("VehId"));
        }


        if (gpsPosition.getVersion().equalsIgnoreCase("2019")) {

        } else {

            for (JTT808_AddtionalInfo addtionalInfo : gpsPosition.getGnssDataAdditionalInfo()) {

                if (addtionalInfo.getID() == 0x65) {
                    JTT808_0x0200_0x65 jtt808_0x0200_0x65 = (JTT808_0x0200_0x65) addtionalInfo;

                    String flagStateStr = jtt808_0x0200_0x65.FlagState == 0 ? "报警开始" : jtt808_0x0200_0x65.FlagState == 1 ? "报警开始" : "报警结束";
                    String alarmLevel = jtt808_0x0200_0x65.AlarmLevel == 1 ? "一级报警" : "二级报警";
                    TCmd020065 tCmd020065 = new TCmd020065();
                    tCmd020065.setAlarmId(Integer.parseUnsignedInt(String.valueOf(jtt808_0x0200_0x65.AlarmId)));
                    tCmd020065.setFlagState(flagStateStr);
                    tCmd020065.setAlarmType(jtt808_0x0200_0x65.AlarmOrEventTypeString);
                    tCmd020065.setAlarmlevel(alarmLevel);
                    tCmd020065.setFatigueDegree(Byte.toUnsignedInt(jtt808_0x0200_0x65.getFatigue()));
                    tCmd020065.setVelocity(jtt808_0x0200_0x65.getSpeed() != 0 ? Byte.toUnsignedInt(jtt808_0x0200_0x65.getSpeed()) : gpsPosition.getSpeedExt());
                    tCmd020065.setHigh(Double.valueOf(Integer.parseUnsignedInt(String.valueOf(jtt808_0x0200_0x65.Altitude))));
                    tCmd020065.setLat(PointUtils.gps2double(Integer.parseUnsignedInt(String.valueOf(jtt808_0x0200_0x65.Latitude))));
                    tCmd020065.setLon(PointUtils.gps2double(Integer.parseUnsignedInt(String.valueOf(jtt808_0x0200_0x65.Longitude))));
                    tCmd020065.setGpsTime(jtt808_0x0200_0x65.AlarmTime);
                    tCmd020065.setVehStatus("");
                    tCmd020065.setVehicleStatus(0);
                    tCmd020065.setState(Byte.toUnsignedInt(jtt808_0x0200_0x65.FlagState));
                    tCmd020065.setAlarmIdNumber(jtt808_0x0200_0x65.AlarmIdentificationStr);
                    tCmd020065.setASTerminalId(jtt808_0x0200_0x65.AlarmIdentification.TerminalID);
                    tCmd020065.setATime(jtt808_0x0200_0x65.AlarmIdentification.Time);
                    tCmd020065.setASerialNumber(String.valueOf(Byte.toUnsignedInt(jtt808_0x0200_0x65.AlarmIdentification.SN)));
                    tCmd020065.setAAppendagNumber(Byte.toUnsignedInt(jtt808_0x0200_0x65.AlarmIdentification.AttachCount));
                    tCmd020065.setVehId(gpsPosition.getVehId());
                    tCmd020065.setDeptId(gpsPosition.getDeptId());
                    tCmd020065.setAlarmNumber(gpsPosition.getJtt808_0x65_platformId());
                    tCmd020065.setSystime(new Date());

                    String driverNo = null;
                    String driverName = null;
                    if (driverlogininfo != null) {
                        driverNo = driverlogininfo.getDriverNo();
                        driverName = driverlogininfo.getDriverName();
                    }

                    tCmd020065.setDriverCode(driverNo);
                    tCmd020065.setDriverName(driverName);


                    collector.collect(tCmd020065);


                }


            }


        }
    }
}